package com.distributer.delayqueue.delayqueue.impl;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.Set;

import org.apache.commons.lang.StringUtils;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.Assert;

import com.distributer.delayqueue.delayqueue.JobOperationService;
import com.distributer.delayqueue.delayqueue.bucket.BucketTask;
import com.distributer.delayqueue.delayqueue.support.RedisQueueProperties;
import com.distributer.delayqueue.delayqueue.support.RedisSupport;
import com.distributer.delayqueue.job.JobMsg;
import com.distributer.delayqueue.job.JobStatus;
import com.distributer.delayqueue.job.JobWrapp;
import com.distributer.delayqueue.util.FastJsonConvert;
import com.distributer.delayqueue.util.NamedUtil;
import com.google.common.collect.Lists;

public class JobOperationServiceImpl implements JobOperationService {

	private RedisSupport redisSupport;
	private RedisQueueProperties redisQueueProperties;

	public void setRedisSupport(RedisSupport redisSupport) {
		this.redisSupport = redisSupport;
	}

	public void setRedisQueueProperties(RedisQueueProperties redisQueueProperties) {
		this.redisQueueProperties = redisQueueProperties;
	}

	/**
	 * 任务池名称
	 * 
	 * @return
	 */
	private String getPoolName() {
		String name = NamedUtil.buildRealTimeName(redisQueueProperties.getProfix(), redisQueueProperties.getName(), redisQueueProperties.getOriginPool());
		return name;
	}

	/**
	 * 待执行队列名称
	 * 
	 * @return
	 */
	private String getReadyName() {
		String name = NamedUtil.buildRealTimeName(redisQueueProperties.getProfix(), redisQueueProperties.getName(), redisQueueProperties.getReadyName());
		return name;
	}

	@Override
	public JobMsg getJob(String jobId) {
		String json = redisSupport.getHashKey(getPoolName(), jobId);
		if (!StringUtils.isEmpty(json)) {
			return FastJsonConvert.convertJSONToObject(json, JobWrapp.class);
		}
		return null;
	}

	@Override
	public void addJobToPool(JobMsg jobMsg) {
		Assert.notNull(jobMsg, "任务为空！");
		redisSupport.hashPut(getPoolName(), jobMsg.getJobId(), FastJsonConvert.convertObjectToJSON(jobMsg));
	}

	@Override
	public void removeJobToPool(String jobId) {
		redisSupport.deleteHashKeys(getPoolName(), jobId);
	}

	@Override
	public void updateJobStatus(String jobId, JobStatus jobSatatus) {
		JobMsg job = getJob(jobId);
		Assert.notNull(job, String.format("根据jobId：%s 查询Job为空！", jobId));
		job.setStatus(jobSatatus.ordinal());
		addJobToPool(job);
	}

	@Override
	public void deleteJobToPool(String jobId) {
		redisSupport.deleteHashKeys(getPoolName(), jobId);
	}

	@Override
	public void addBucketJob(String bucketName, String jobId, double score) {
		redisSupport.zadd(bucketName, jobId, score);
	}

	@Override
	public void removeBucketKey(String bucketName, String jobId) {
		redisSupport.zrem(bucketName, jobId);
	}

	@Override
	public void addReadyTime(String readyName, String jobId) {
		redisSupport.rightPush(getReadyName(), jobId);
	}

	@Override
	public String getReadyJob() {
		return redisSupport.leftPop(getReadyName());
	}

	@Override
	public List<String> getReadyJob(int size) {
		List<String> jobIds = redisSupport.lrange(getReadyName(), 0, size);
		if (null != jobIds && jobIds.size() > 0) {
			jobIds = Lists.reverse(jobIds);
		}
		return jobIds;
	}

	@Override
	public boolean removeReadyJob(String jobId) {
		return redisSupport.lrem(getReadyName(), jobId);
	}

	@Override
	public String getBucketTop1Job(String bucketName) {
		double to = Long.valueOf(System.currentTimeMillis() + BucketTask.TIME_OUT).doubleValue();
		Set<String> sets = redisSupport.zrangeByScore(bucketName, 0, to, 0, 1);
		if (sets != null && sets.size() > 0) {
			String jobMsgId = Objects.toString(sets.toArray()[0]);
			return jobMsgId;
		}
		return null;
	}

	@Override
	public List<String> getBucketTopJobs(String bucketName, int size) {
		double to = Long.valueOf(System.currentTimeMillis() + BucketTask.TIME_OUT).doubleValue();
		Set<ZSetOperations.TypedTuple<String>> sets = redisSupport.zrangeByScoreWithScores(bucketName, 0, to, 0, size);
		List<String> lsts = Lists.newArrayList();
		if (sets != null && sets.size() > 0) {
			Iterator<ZSetOperations.TypedTuple<String>> it = sets.iterator();
			while (it.hasNext()) {
				ZSetOperations.TypedTuple<String> curr = it.next();
				if (curr.getScore() <= System.currentTimeMillis()) {
					lsts.add(curr.getValue());
				} else {
					break;
				}
			}
			return lsts;
		}
		return null;
	}

	@Override
	public void clearAll() {
		int bucketSize = redisQueueProperties.getBucketSize();
		if (bucketSize <=0) {
			bucketSize = 1;
		}
		List<String> keys = new ArrayList<String>();
		for (int i = 0; i < bucketSize; i++) {
			keys.add(NamedUtil.buildBucketName(redisQueueProperties.getProfix(), redisQueueProperties.getName(), i));
		}
		keys.add(getPoolName());
		keys.add(getReadyName());
		redisSupport.deleteKey(keys.toArray(new String[]{}));
	}

}
